//! # 消息总线核心功能单元测试
//!
//! 全面测试消息总线系统的核心功能，确保系统可靠性和稳定性

use rustcloud_bus::{
    bus::{MessageBus},
    memory::InMemoryMessageBus,
    message::{Message, MessageType, MessagePriority},
    error::{MessageBusError, MessageBusResult},
    event::{Event, EventType, EventContext, EventBus},
    event_bus::MessageBusEventBus,
    monitoring::{InMemoryMonitor, MonitoringConfig, Monitor},
    admin_api::{HttpAdminApi, AdminApiConfig, AdminApi},
    dead_letter::{InMemoryDeadLetterQueue, DeadLetterConfig, DeadLetterQueue},
    file_storage::{FileStorage, FileStorageConfig},
    persistence::{MessageStorage, MessageStatus, QueryCriteria},
};
use serde_json::json;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::timeout;
use tempfile::TempDir;

/// 测试消息总线基本功能
#[tokio::test]
async fn test_message_bus_basic_operations() {
    let bus = InMemoryMessageBus::new().await.unwrap();

    // 测试发布消息（无订阅者）
    let message = Message::new("test.topic", json!({"data": "test"}));
    assert!(bus.publish(message).await.is_ok());

    // 测试统计信息
    let stats = bus.stats().await.unwrap();
    assert_eq!(stats.total_published, 1);
    assert_eq!(stats.active_subscribers, 0);

    // 测试关闭
    assert!(bus.close().await.is_ok());
}

/// 测试发布订阅模式
#[tokio::test]
async fn test_publish_subscribe_pattern() {
    let bus = InMemoryMessageBus::new().await.unwrap();

    // 创建订阅者
    let subscriber = bus.subscribe("test.events.*").await.unwrap();

    // 发布匹配消息
    let message1 = Message::new("test.events.created", json!({"id": 1}));
    bus.publish(message1).await.unwrap();

    // 发布不匹配消息
    let message2 = Message::new("other.topic", json!({"id": 2}));
    bus.publish(message2).await.unwrap();

    // 接收消息
    let received = subscriber.receive_timeout(1000).await.unwrap();
    assert!(received.is_some());
    
    let msg = received.unwrap();
    assert_eq!(msg.topic(), "test.events.created");
    assert_eq!(msg.payload()["id"], 1);

    // 验证不匹配的消息没有收到
    let no_message = subscriber.receive_timeout(100).await.unwrap();
    assert!(no_message.is_none());
}

/// 测试多个订阅者
#[tokio::test]
async fn test_multiple_subscribers() {
    let bus = InMemoryMessageBus::new().await.unwrap();

    // 创建多个订阅者
    let subscriber1 = bus.subscribe("broadcast.*").await.unwrap();
    let subscriber2 = bus.subscribe("broadcast.*").await.unwrap();
    let subscriber3 = bus.subscribe("other.*").await.unwrap();

    // 发布广播消息
    let message = Message::new("broadcast.news", json!({"content": "重要消息"}));
    bus.publish(message).await.unwrap();

    // 验证前两个订阅者都收到消息
    let received1 = subscriber1.receive_timeout(1000).await.unwrap();
    let received2 = subscriber2.receive_timeout(1000).await.unwrap();
    let received3 = subscriber3.receive_timeout(100).await.unwrap();

    assert!(received1.is_some());
    assert!(received2.is_some());
    assert!(received3.is_none()); // 第三个订阅者不应该收到

    // 验证消息内容
    assert_eq!(received1.unwrap().payload()["content"], "重要消息");
    assert_eq!(received2.unwrap().payload()["content"], "重要消息");
}

/// 测试主题模式匹配
#[tokio::test]
async fn test_topic_pattern_matching() {
    let bus = InMemoryMessageBus::new().await.unwrap();

    // 测试通配符订阅
    let subscriber = bus.subscribe("user.events.*").await.unwrap();

    // 测试各种主题
    let test_cases = vec![
        ("user.events.created", true),
        ("user.events.updated", true), 
        ("user.events.deleted", true),
        ("user.profile.updated", false),
        ("admin.events.created", false),
        ("user.events", false),
    ];

    for (topic, should_receive) in test_cases {
        let message = Message::new(topic, json!({"test": topic}));
        bus.publish(message).await.unwrap();

        let received = subscriber.receive_timeout(100).await.unwrap();
        if should_receive {
            assert!(received.is_some(), "应该收到消息: {}", topic);
            assert_eq!(received.unwrap().topic(), topic);
        } else {
            assert!(received.is_none(), "不应该收到消息: {}", topic);
        }
    }
}

/// 测试消息验证
#[tokio::test]
async fn test_message_validation() {
    let bus = InMemoryMessageBus::new().await.unwrap();

    // 测试空主题
    let mut empty_topic_message = Message::new("", json!({}));
    empty_topic_message.metadata.topic = String::new();
    
    let result = bus.publish(empty_topic_message).await;
    assert!(result.is_err());

    // 测试过大消息
    let large_data = "x".repeat(2_000_000); // 2MB
    let large_message = Message::new("test.large", json!({"data": large_data}));
    
    let result = bus.publish(large_message).await;
    assert!(result.is_err());
}

/// 测试订阅管理
#[tokio::test]
async fn test_subscription_management() {
    let bus = InMemoryMessageBus::new().await.unwrap();

    // 创建订阅
    let subscriber = bus.subscribe("test.*").await.unwrap();
    let subscription_id = subscriber.subscription_id().to_string();

    // 验证订阅者计数
    let count = bus.subscriber_count("test.topic").await.unwrap();
    assert_eq!(count, 1);

    // 取消订阅
    bus.unsubscribe(&subscription_id).await.unwrap();

    // 验证订阅者计数
    let count = bus.subscriber_count("test.topic").await.unwrap();
    assert_eq!(count, 0);

    // 重复取消订阅应该失败
    let result = bus.unsubscribe(&subscription_id).await;
    assert!(result.is_err());
}

/// 测试事件总线
#[tokio::test]
async fn test_event_bus() {
    let message_bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    let event_bus = MessageBusEventBus::new(message_bus).await.unwrap();

    // 创建事件
    let context = EventContext::new("test-service".to_string())
        .with_user_id("user123".to_string());
    
    let event = Event::new(
        "user.registered".to_string(),
        EventType::Domain,
        json!({"user_id": "user123", "email": "test@example.com"}),
        context,
    );

    // 发布事件
    let result = event_bus.publish_event(event).await;
    assert!(result.is_ok());
}

/// 测试消息存储（文件存储）
#[tokio::test]
async fn test_file_storage() {
    let temp_dir = TempDir::new().unwrap();
    
    let config = FileStorageConfig {
        storage_dir: temp_dir.path().to_path_buf(),
        enable_compression: false,
        shard_size: 1024 * 1024, // 1MB
        index_cache_size: 100,
        cleanup_interval_secs: 60,
    };

    let storage = FileStorage::new(config).await.unwrap();

    // 测试存储消息
    let message = Message::new("test.storage", json!({"data": "test_data"}));
    let message_id = storage.store(message.clone()).await.unwrap();

    // 测试获取消息
    let retrieved = storage.get(&message_id).await.unwrap();
    assert!(retrieved.is_some());
    
    let record = retrieved.unwrap();
    assert_eq!(record.message.topic(), "test.storage");
    assert_eq!(record.message.payload()["data"], "test_data");
    assert_eq!(record.status, MessageStatus::Pending);

    // 测试更新状态
    storage.update_status(&message_id, MessageStatus::Processing).await.unwrap();
    
    let updated = storage.get(&message_id).await.unwrap().unwrap();
    assert_eq!(updated.status, MessageStatus::Processing);

    // 测试查询
    let criteria = QueryCriteria {
        status: Some(MessageStatus::Processing),
        ..Default::default()
    };
    let results = storage.query(criteria).await.unwrap();
    assert_eq!(results.len(), 1);
}

/// 测试死信队列
#[tokio::test]
async fn test_dead_letter_queue() {
    let config = DeadLetterConfig::default();
    let dlq = InMemoryDeadLetterQueue::new(config);

    // 添加死信消息
    let message = Message::new("test.failed", json!({"error": "test"}));
    let dl_id = dlq.add_dead_letter(
        message,
        "处理失败".to_string(),
        None,
    ).await.unwrap();

    // 获取死信消息
    let retrieved = dlq.get_dead_letter(&dl_id).await.unwrap();
    assert!(retrieved.is_some());
    
    let dl_msg = retrieved.unwrap();
    assert_eq!(dl_msg.failure_reason, "处理失败");
    assert_eq!(dl_msg.retry_count, 0);

    // 测试重试
    let retry_result = dlq.retry_message(&dl_id).await.unwrap();
    assert!(retry_result);

    // 标记重试成功
    dlq.mark_retry_success(&dl_id).await.unwrap();

    // 验证消息已被移除
    let removed = dlq.get_dead_letter(&dl_id).await.unwrap();
    assert!(removed.is_none());
}

/// 测试监控系统
#[tokio::test]
async fn test_monitoring_system() {
    let config = MonitoringConfig::default();
    let monitor = InMemoryMonitor::new(config);

    // 记录一些指标
    monitor.record_message_published("test.topic", Duration::from_millis(10)).await;
    monitor.record_message_processed("test.topic", true, Duration::from_millis(20)).await;
    monitor.record_error("test_error", &MessageBusError::internal_error("test error")).await;

    // 获取指标
    let metrics = monitor.get_metrics().await.unwrap();
    assert!(metrics.throughput.total_messages > 0);
    assert!(metrics.error_rate.total_errors > 0);

    // 执行健康检查
    let health = monitor.health_check().await.unwrap();
    assert_eq!(health.components.len(), 0); // 没有注册组件
}

/// 测试管理API
#[tokio::test]
async fn test_admin_api() {
    let message_bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    let monitor = Arc::new(InMemoryMonitor::new(MonitoringConfig::default()));
    let config = AdminApiConfig::default();
    
    let admin_api = HttpAdminApi::new(config, message_bus, monitor);

    // 测试系统信息
    let system_info = admin_api.get_system_info().await.unwrap();
    assert!(!system_info.version.is_empty());
    
    // 测试发布消息
    let publish_request = rustcloud_bus::admin_api::PublishMessageRequest {
        topic: "api.test".to_string(),
        payload: json!({"test": "data"}),
        headers: None,
    };
    
    let message_id = admin_api.publish_message(publish_request).await.unwrap();
    assert!(!message_id.is_empty());
}

/// 测试并发安全性
#[tokio::test]
async fn test_concurrent_operations() {
    let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    
    // 创建多个并发任务
    let mut tasks = Vec::new();
    
    for i in 0..10 {
        let bus_clone = bus.clone();
        let task = tokio::spawn(async move {
            // 并发发布消息
            for j in 0..10 {
                let message = Message::new(
                    &format!("concurrent.test.{}.{}", i, j),
                    json!({"thread": i, "message": j})
                );
                bus_clone.publish(message).await.unwrap();
            }
        });
        tasks.push(task);
    }
    
    // 等待所有任务完成
    for task in tasks {
        task.await.unwrap();
    }
    
    // 验证统计信息
    let stats = bus.stats().await.unwrap();
    assert_eq!(stats.total_published, 100);
}

/// 测试错误处理
#[tokio::test]
async fn test_error_handling() {
    let bus = InMemoryMessageBus::new().await.unwrap();

    // 测试在关闭的总线上发布消息
    bus.close().await.unwrap();
    
    let message = Message::new("test.closed", json!({}));
    let result = bus.publish(message).await;
    assert!(result.is_err());
    
    // 测试在关闭的总线上订阅
    let result = bus.subscribe("test.*").await;
    assert!(result.is_err());
}

/// 测试内存压力
#[tokio::test]
async fn test_memory_pressure() {
    let bus = InMemoryMessageBus::new().await.unwrap();
    let subscriber = bus.subscribe("stress.test.*").await.unwrap();

    // 发布大量小消息
    for i in 0..1000 {
        let message = Message::new(
            &format!("stress.test.{}", i),
            json!({"index": i, "data": "small_message"})
        );
        
        let result = timeout(
            Duration::from_millis(100),
            bus.publish(message)
        ).await;
        
        match result {
            Ok(Ok(())) => {},
            Ok(Err(e)) => panic!("发布消息失败: {}", e),
            Err(_) => panic!("发布消息超时"),
        }
    }

    // 验证统计信息
    let stats = bus.stats().await.unwrap();
    assert_eq!(stats.total_published, 1000);
    
    // 快速消费一些消息以避免内存堆积
    for _ in 0..100 {
        let _ = subscriber.receive_timeout(10).await;
    }
}

/// 测试消息优先级
#[tokio::test]
async fn test_message_priority() {
    // 创建不同优先级的消息
    let high_priority_msg = Message::new("test.priority", json!({"data": "high"}))
        .with_metadata(
            rustcloud_bus::message::MessageMetadata::new("test.priority")
                .with_priority(MessagePriority::High)
        );
    
    let low_priority_msg = Message::new("test.priority", json!({"data": "low"}))
        .with_metadata(
            rustcloud_bus::message::MessageMetadata::new("test.priority")
                .with_priority(MessagePriority::Low)
        );

    // 验证优先级设置
    assert_eq!(high_priority_msg.metadata().attributes.priority, MessagePriority::High);
    assert_eq!(low_priority_msg.metadata().attributes.priority, MessagePriority::Low);
}

/// 测试消息类型
#[tokio::test]
async fn test_message_types() {
    // 测试不同类型的消息创建
    let event_msg = Message::event("user.created", json!({"user_id": 123}));
    let command_msg = Message::command("user.delete", json!({"user_id": 123}));
    let query_msg = Message::query("user.get", json!({"user_id": 123}));
    let notification_msg = Message::notification("email.sent", json!({"to": "user@example.com"}));

    // 验证消息类型
    assert_eq!(event_msg.metadata().attributes.message_type, MessageType::Event);
    assert_eq!(command_msg.metadata().attributes.message_type, MessageType::Command);
    assert_eq!(query_msg.metadata().attributes.message_type, MessageType::Query);
    assert_eq!(notification_msg.metadata().attributes.message_type, MessageType::Notification);
}

/// 集成测试 - 完整流程
#[tokio::test]
async fn test_integration_full_workflow() {
    // 创建所有组件
    let message_bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    let monitor = Arc::new(InMemoryMonitor::new(MonitoringConfig::default()));
    let event_bus = MessageBusEventBus::new(message_bus.clone()).await.unwrap();
    
    // 启动监控
    monitor.start().await.unwrap();
    
    // 创建订阅者
    let subscriber = message_bus.subscribe("integration.test.*").await.unwrap();
    
    // 发布事件
    let context = EventContext::new("integration-test".to_string());
    let event = Event::new(
        "integration.test.started".to_string(),
        EventType::Domain,
        json!({"test_id": "integration_001", "timestamp": chrono::Utc::now()}),
        context,
    );
    
    event_bus.publish_event(event).await.unwrap();
    
    // 接收并处理消息
    let received = subscriber.receive_timeout(1000).await.unwrap();
    assert!(received.is_some());
    
    let message = received.unwrap();
    assert!(message.topic().starts_with("events."));
    
    // 记录处理指标
    monitor.record_message_processed(message.topic(), true, Duration::from_millis(50)).await;
    
    // 验证指标
    let metrics = monitor.get_metrics().await.unwrap();
    assert!(metrics.throughput.successful_messages > 0);
    
    // 获取系统统计
    let stats = message_bus.stats().await.unwrap();
    assert!(stats.total_published > 0);
    assert!(stats.active_subscribers > 0);
}

/// 运行所有测试的辅助函数
pub async fn run_all_tests() -> MessageBusResult<()> {
    println!("🧪 开始运行消息总线核心功能测试...\n");
    
    let mut passed = 0;
    let mut failed = 0;
    
    // 由于测试函数的复杂性，我们采用手动执行的方式
    println!("  • 测试 消息总线基本操作: ");
    match test_message_bus_basic_operations().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 发布订阅模式: ");
    match test_publish_subscribe_pattern().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 多订阅者: ");
    match test_multiple_subscribers().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 主题模式匹配: ");
    match test_topic_pattern_matching().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 消息验证: ");
    match test_message_validation().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 订阅管理: ");
    match test_subscription_management().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 事件总线: ");
    match test_event_bus().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 文件存储: ");
    match test_file_storage().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 死信队列: ");
    match test_dead_letter_queue().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 监控系统: ");
    match test_monitoring_system().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 管理API: ");
    match test_admin_api().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 并发操作: ");
    match test_concurrent_operations().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 错误处理: ");
    match test_error_handling().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 内存压力: ");
    match test_memory_pressure().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 消息优先级: ");
    match test_message_priority().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 消息类型: ");
    match test_message_types().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }

    println!("  • 测试 集成测试: ");
    match test_integration_full_workflow().await {
        Ok(()) => {
            println!("✅ 通过");
            passed += 1;
        }
        Err(e) => {
            println!("❌ 失败 - {:?}", e);
            failed += 1;
        }
    }
    
    println!("\n📊 测试结果统计:");
    println!("  • 通过: {} 个", passed);
    println!("  • 失败: {} 个", failed);
    println!("  • 总计: {} 个", passed + failed);
    println!("  • 成功率: {:.1}%", (passed as f64 / (passed + failed) as f64) * 100.0);
    
    if failed == 0 {
        println!("\n🎉 所有测试通过！");
        Ok(())
    } else {
        Err(MessageBusError::internal_error(format!("测试失败: {} 个", failed)))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn run_test_suite() {
        run_all_tests().await.unwrap();
    }
}